package com.peng.sparktest.sparkstreaming

import java.io.{BufferedReader, InputStreamReader}
import java.net.Socket

import org.apache.spark.storage.StorageLevel
import org.apache.spark.streaming.receiver.Receiver

class MySocketReceiver(host: String, port: Int) extends Receiver[String](StorageLevel.DISK_ONLY) {
  override def onStart(): Unit = {
    new Thread() {


      override def run(): Unit = {
        getBatchDataFromSocket()
      }
    }.start()
  }

  private def getBatchDataFromSocket(): Unit = {
    val socket = new Socket(host, port)
    val reader: BufferedReader = new BufferedReader(new InputStreamReader(socket.getInputStream))
    var line: String = reader.readLine()
    while (!isStopped() && line != null) {
      store(line)
      line = reader.readLine()
    }
  }

  override def onStop(): Unit = ???
}
